package org.apache.flink.client.cli;

import java.net.InetSocketAddress;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.util.ERException;
import org.apache.flink.util.NetUtils;

/**
 * @类名: DefaultCLI
 * @说明: 默认的CLI，用于与standalone集群交互
 *
 * @author   leehom
 * @Date	 2024年2月25日 下午3:26:09
 * 修改记录：
 *
 * @see 	 
 */
public class DefaultCLI extends AbstractCustomCommandLine {

    public static final String ID = "default";

    private static final Option addressOption =
            new Option(
                    "m",
                    "resourceconsumer",
                    true,
                    "Address of the resourceconsumer to which to connect. "
                            + "Use this flag to connect to a different resourceconsumer than the one specified in the configuration. "
                            + "Attention: This option is respected only if the high-availability configuration is NONE.");

    @Override
    public boolean isActive(CommandLine commandLine) {
        // always active because we can try to read a JobManager address from the config
        return true;
    }

    @Override
    public Configuration toConfiguration(CommandLine commandLine) throws ERException {

        final Configuration resultingConfiguration = super.toConfiguration(commandLine);
        if (commandLine.hasOption(addressOption.getOpt())) {
            String addressWithPort = commandLine.getOptionValue(addressOption.getOpt());
            InetSocketAddress jobManagerAddress = NetUtils.parseHostPortAddress(addressWithPort);
            setJobManagerAddressInConfig(resultingConfiguration, jobManagerAddress);
        }
        resultingConfiguration.setString(DeploymentOptions.TARGET, "remote");

        DynamicPropertiesUtil.encodeDynamicProperties(commandLine, resultingConfiguration);

        return resultingConfiguration;
    }

    static void setJobManagerAddressInConfig(Configuration config, InetSocketAddress address) {
        config.setString(ConsumerManagerOptions.ADDRESS, address.getHostString());
        config.setInteger(ConsumerManagerOptions.PORT, address.getPort());
        config.setString(RestOptions.ADDRESS, address.getHostString());
        config.setInteger(RestOptions.PORT, address.getPort());
    }
    
    @Override
    public String getId() {
        return ID;
    }

    @Override
    public void addGeneralOptions(Options baseOptions) {
        super.addGeneralOptions(baseOptions);
        baseOptions.addOption(addressOption);
        baseOptions.addOption(DynamicPropertiesUtil.DYNAMIC_PROPERTIES);
    }
}
